 1.Flink下沉数据到kudu
   
   思路：
   自定义下沉器 ： extends RichSinkFunction
   1).依赖：
    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-client -->
        <dependency>
            <groupId>org.apache.kudu</groupId>
            <artifactId>kudu-client</artifactId>
            <version>1.5.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.11.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.11.1</version>
            <scope>provided</scope>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.11.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-bridge -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.12</artifactId>
            <version>1.11.1</version>
            <scope>provided</scope>
        </dependency>
	</dependencies>

   2).数据源：
new UserInfo("001", "Jack", 18),
new UserInfo("002", "Rose", 20),
new UserInfo("003", "Cris", 22),
new UserInfo("004", "Lily", 19),
new UserInfo("005", "Lucy", 21),
new UserInfo("006", "Json", 24)
   3).自定义下沉器
package com.lagou.sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kudu.client.*;

import java.util.Map;

public class MySinkToKudu extends RichSinkFunction<Map<String, Object>> {

    private String masterAddr;
    private String tableName;

    private KuduClient client;
    private KuduTable table;
    private KuduSession session;

    public MySinkToKudu(String masterAddr, String tableName) {
        this.masterAddr = masterAddr;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new KuduClient.KuduClientBuilder(masterAddr).build();
        table = client.openTable(tableName);
        session = client.newSession();
    }

    //schema id String  name String   age int
    public void invoke(Map<String, Object> value, Context context) throws Exception {
        Insert insert = table.newInsert();
        PartialRow row = insert.getRow();
        row.addString("id", value.get("id").toString());
        row.addString("name", value.get("name").toString());
        row.addInt("age", Integer.valueOf(value.get("age").toString()));
        session.flush();
        session.apply(insert);
        session.close();
    }

    @Override
    public void close() throws Exception {
        client.close();
    }
}

   4).测试
package com.lagou.sink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.Map;

public class FlinkToKuduDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<UserInfo> data = env.fromElements(
                new UserInfo("001", "Jack", 18),
                new UserInfo("002", "Rose", 20),
                new UserInfo("003", "Cris", 22),
                new UserInfo("004", "Lily", 19),
                new UserInfo("005", "Lucy", 21),
                new UserInfo("006", "Json", 24)
        );

        SingleOutputStreamOperator<Map<String, Object>> maped = data.map(new MapFunction<UserInfo, Map<String, Object>>() {
            public Map<String, Object> map(UserInfo value) throws Exception {
                HashMap<String, Object> map = new HashMap<String, Object>();
                map.put("id", value.getId());
                map.put("name", value.getName());
                map.put("age", value.getAge());
                return map;
            }
        });

        String masterAddr = "linux122";
        String tableName = "flinktable";
        maped.addSink(new MySinkToKudu(masterAddr, tableName));

        env.execute();
    }
}
   
   实体类：
package com.lagou.sink;

public class UserInfo {
    private String id;
    private String name;
    private int age;

    public UserInfo(String id, String name, int age) {
        this.id = id;
        this.name = name;
        this.age = age;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "UserInfo{" +
                "id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", age=" + age +
                '}';
    }
}
